package com.bjtu.boatmq.client.producer;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.I0Itec.zkclient.ZkClient;

import com.bjtu.boatmq.common.log.Log;

import cn.boat.router.MetaZookeeper;
import cn.boat.router.broker.BrokerZookeeper;
import cn.boat.router.models.Broker;
import cn.boat.router.models.Partition;
import cn.boat.router.models.Topic;
import cn.boat.router.producer.ProducerZookeeper;
import cn.boat.router.utils.ZkClientUtils;

/**
 * 路由器类
 * @author dell
 *
 */
public class Router {
	private Map<String,Topic> r=new HashMap<String,Topic>();
	private Map<Integer,Broker> b=new HashMap<Integer,Broker>();
	private ZkClientUtils.ZKConfig zkConfig;
	private ZkClient zkClient;
	private BrokerZookeeper bz;
	private ProducerZookeeper pz;
	private Log log;
	public Router(String ip) {
		super();
		log=new Log();
		this.zkConfig= new ZkClientUtils.ZKConfig();
		this.zkConfig.setZkConnect(ip);
		this.zkClient=new ZkClient(this.zkConfig.zkConnect,this.zkConfig.zkSessionTimeoutMs,
				this.zkConfig.zkConnectionTimeoutMs,new ZkClientUtils.StringSerializer());
		this.bz=new BrokerZookeeper(this.zkClient,new MetaZookeeper(this.zkClient,"/boatMQ"));
		this.pz=new ProducerZookeeper(this.zkClient,new MetaZookeeper(this.zkClient,"/boatMQ"));
	}
	//获取指定topic路由信息，包括分区，分区对应master-broker
//	public void getRoute(List<String> topics) {
//		
//	}
	public Topic getRoute(String topic) throws Exception {
		Topic t=bz.getTopicByName(topic);
		if(t==null)
			return null;
		r.put(topic, t);
		for(Partition part:t.getPartitionList()) {
			getBroker(part.getLeader());
		}
		return t;
	}
	public void getRoute() throws Exception {
		Set<String> map=this.r.keySet();
		for(String topic:map) {
			Topic t=this.getRoute(topic);
			if(t==null)
				return;
			for(Partition part:t.getPartitionList()) {
				getBroker(part.getLeader());
			}
		}
	}
	public void getBroker(int id) throws Exception {
		if(this.b.get(id)==null) {
			Broker broker=bz.getBrokerById(id);
			b.put(id,broker);
		}
		
	}
	//创建topic，返回该topic的路由信息
	public void createTopic(String topic,int numOfPartion,int numOfBackups) throws Exception{
		this.r.put(topic, null);
		pz.registerTopic(topic, numOfPartion, numOfBackups);
		this.getRoute(topic);
		return;
	}
	public Topic getTopic(String name) {
		return this.r.get(name);	
	}
	public Broker getBrokerById(int id) throws Exception {
		this.getBroker(id);
		return this.b.get(id);
	}
	
	
}
